package com.qingcloud.df.executor.eventbus;

import akka.actor.AbstractActor;
import com.qingcloud.df.executor.enums.ComponentCommunicationEnum;
import com.qingcloud.df.executor.launcher.process.ComponentExecutorHandle;
import com.qingcloud.df.executor.queue.ComponentMessage;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import com.qingcloud.df.sdk.component.message.ComponentValue;
import com.qingcloud.df.sdk.component.transport.akka.AkkaTransporter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.alibaba.fastjson.JSON.parseObject;

/**
 * 事件订阅actor
 * @author yangfan
 * @date 2022/1/17 10:45
 */
public class MessageEventSubActor extends AbstractActor {

    Logger logger =  LoggerFactory.getLogger(MessageEventSubActor.class);

    private final CompInstContext context;

    public MessageEventSubActor(CompInstContext context){
        this.context = context;
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder().matchAny(msg -> {
            MessageEvent messageEvent = (MessageEvent) msg;
            try {
                ComponentValue componentValue = parseObject(messageEvent.getMessage(), ComponentValue.class);
                ComponentMessage componentMessage = new ComponentMessage();
                componentMessage.setTopic(messageEvent.getTopic());
                componentMessage.setComponentValue(componentValue);
                componentMessage.setCommunicationType(ComponentCommunicationEnum.AKKA.getCommunicationType());
                Integer senderId = componentValue.getSenderId();
                componentMessage.setSendInstanceId(senderId);
                componentMessage.setInstanceId(context.getId());
                componentMessage.setMessageFrom(messageEvent.getMessageFrom());
                ComponentExecutorHandle.fusion(componentMessage);
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("----------akka receiver error--------");
            }
        }).build();
    }

    @Override
    public void postStop() throws Exception {
        super.postStop();
    }
}
